package twitterhadoop.app;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

import twitterhadoop.hadoop.io.TextArrayWritable;

public class Map extends Mapper<LongWritable, TextArrayWritable, ClosedItemSet, TextArrayWritable> {
	private static final String[] EMPTY_STRING_ARRAY = new String[0];
	private static final Log LOG = LogFactory.getLog(Map.class);

	List<ClosedItemSet> closedItemsets = new ArrayList<ClosedItemSet>();

	@Override
	/**
	 * Reads the current closed item-sets from the distributed cache.
	 */
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		Configuration conf = context.getConfiguration();
		Path[] localFiles = DistributedCache.getLocalCacheFiles(conf);
		if (localFiles == null) {
			closedItemsets.add(new ClosedItemSet(EMPTY_STRING_ARRAY, EMPTY_STRING_ARRAY, 0));
			return;
		}
		
		for (Path closedItemsetsPath : localFiles) {
			BufferedReader fis = new BufferedReader(new FileReader(closedItemsetsPath.toString()));
			String line;
			while ((line = fis.readLine()) != null) {
				closedItemsets.add(ClosedItemSet.deserialize(line));
			}
			fis.close();			
		}

		/*
		Path closedItemsetsPath = localFiles[0];

		BufferedReader fis = new BufferedReader(new FileReader(closedItemsetsPath.toString()));
		String line;
		while ((line = fis.readLine()) != null) {
			closedItemsets.add(ClosedItemSet.deserialize(line));
		}
		fis.close();
		*/
	}

	/**
	 * 
	 */
	public void map(LongWritable key, TextArrayWritable value, Context context) throws IOException, InterruptedException {
		//TextArrayWritable outKey = new TextArrayWritable();
		
		String[] valueArray = value.toArray();
		
		TextArrayWritable outValue = new TextArrayWritable();
		outValue.setArray(valueArray);

		for (ClosedItemSet projector : this.closedItemsets) {
			String[] projected = valueArray;

			// Removing the projector from the items.
			// We must check is the projected contains all the projector.			 
			List<String> list = new ArrayList<String>(Arrays.asList(projected));
			int projectedLengthBeforeRemove = list.size();			
			list.removeAll(Arrays.asList(projector.getItemset()));
			
			if (projectedLengthBeforeRemove != list.size() + projector.getItemset().length) {
				// the projected does not contain the projector
				continue;
			}
			
			// YGYG toArray(new String[v2.size()]);
			// List<String> temp = new ArrayList<String>(list);
			// temp.addAll(Arrays.asList(projector.getGenerator()));
			// Collections.sort(temp);
			// String[] stringOutValue = temp.toArray(new String[list.size()]);
			// outValue.setArray(stringOutValue);
			
			projected = list.toArray(EMPTY_STRING_ARRAY);

			for (int i = 0; i < projected.length; i++) {
				String item = projected[i];
				// Adding item to the projector
				String lastItemOfGenerator = new String();
				int generatorLength = projector.getGenerator().length;
				if (generatorLength > 0) {
					lastItemOfGenerator = projector.getGenerator()[generatorLength - 1]; 
				}
				
				if (item.compareTo(lastItemOfGenerator) > 0) {
				//if (Reduce.compareArrays(new String[] {item}, projector.getGenerator()) > 0) {
					// YGYG trying to add to the generator and not the closure.
					String[] cisWithNewItem       = (String[]) ArrayUtils.add(projector.getItemset(), item);
					String[] generatorWithNewItem = (String[]) ArrayUtils.add(projector.getGenerator(), item);
					Arrays.sort(cisWithNewItem);
					Arrays.sort(generatorWithNewItem);
					//outKey.setArray(outKeyArray);
					ClosedItemSet outKey = new ClosedItemSet(generatorWithNewItem, cisWithNewItem, item, 0);
						
					context.write(outKey, outValue);					
				}
			}
		}
	}
}
